package consume_async

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/segmentio/kafka-go"
	"go-demo-2025/common"
	"go-demo-2025/utils/kafkaUtil"
	"log"
	"sync"
	"time"
)

// 配置参数
const (
	batchSize    = 100             // 批量提交阈值
	batchTimeout = 5 * time.Second // 批量提交超时时间
	workerCount  = 5               // 并发工作协程数
	msgChanSize  = 200             // 消息通道缓冲大小
)

// OrderMessage 订单消息结构体（与同步消费保持一致）
type OrderMessage struct {
	OrderID string `json:"order_id"`
	Amount  int64  `json:"amount"`
	Status  string `json:"status"`
}

// 批量消息处理器
type AsyncConsumer struct {
	reader  *kafka.Reader
	msgChan chan kafka.Message // 消息缓冲通道
	wg      sync.WaitGroup
	ctx     context.Context
	cancel  context.CancelFunc
}

// 创建异步消费者
func NewAsyncConsumer(topic string) *AsyncConsumer {
	ctx, cancel := context.WithCancel(context.Background())
	reader := kafka.NewReader(kafka.ReaderConfig{
		Brokers:     kafkaUtil.KafkaBrokers,
		Topic:       topic,
		Partition:   0, // 固定消费第0分区（测试/单分区场景稳定）
		GroupID:     "order_async_consumer_group",
		MinBytes:    0,                // 有消息立即返回，避免批量阻塞
		MaxBytes:    10 * 1024 * 1024, // 单次拉取最大字节数
		StartOffset: kafka.FirstOffset,
	})

	return &AsyncConsumer{
		reader:  reader,
		msgChan: make(chan kafka.Message, msgChanSize), // 缓冲通道避免阻塞
		ctx:     ctx,
		cancel:  cancel,
	}
}

// 启动消费者
func (c *AsyncConsumer) Start() error {
	log.Println(" 异步消费者启动成功！")
	log.Printf(" 消费配置：主题=%s, 分区=0, 消费者组=%s", "order_events", "order_async_consumer_group")
	log.Printf(" 并发配置：工作协程数=%d, 批量阈值=%d条, 超时时间=%v", workerCount, batchSize, batchTimeout)
	log.Printf(" 等待接收新消息...\n")

	// 启动工作协程（处理消息+批量提交）
	for i := 0; i < workerCount; i++ {
		c.wg.Add(1)
		go c.worker(i)
	}

	// 启动单独的拉取协程（循环拉取消息到通道）
	go c.fetchMessages()

	return nil
}

// 循环拉取消息，存入缓冲通道（模拟批量效果）
func (c *AsyncConsumer) fetchMessages() {
	defer close(c.msgChan) // 退出时关闭通道，通知工作协程
	for {
		select {
		case <-c.ctx.Done():
			log.Println("\n 消息拉取协程收到退出信号，正在关闭...")
			return
		default:
			log.Printf(" 拉取协程：正在拉取消息...")
			// 单条拉取（segmentio/kafka-go 原生方法）
			msg, err := c.reader.ReadMessage(c.ctx)
			if err != nil {
				log.Printf(" 拉取协程：拉取消息失败: %v，1秒后重试...\n", err)
				time.Sleep(1 * time.Second)
				continue
			}

			// 非阻塞写入通道
			select {
			case c.msgChan <- msg:
				log.Printf(" 拉取协程：消息入队成功")
				log.Printf("   偏移量: %d, 消息Key: %s, 消息内容: %s\n", msg.Offset, string(msg.Key), string(msg.Value))
			case <-c.ctx.Done():
				return
			}
		}
	}
}

// 工作协程：处理消息并批量提交
func (c *AsyncConsumer) worker(id int) {
	defer c.wg.Done()
	batch := make([]kafka.Message, 0, batchSize) // 批量提交队列
	ticker := time.NewTicker(batchTimeout)       // 批量超时定时器
	defer ticker.Stop()

	log.Printf(" 工作协程%d：启动成功，开始处理消息", id)

	for {
		select {
		case <-c.ctx.Done():
			log.Printf("\n 工作协程%d：收到退出信号", id)
			// 提交剩余未提交的消息
			if len(batch) > 0 {
				log.Printf("工作协程%d：退出前提交剩余%d条消息", id, len(batch))
				c.commitBatch(id, batch)
			}
			return

		case msg, ok := <-c.msgChan:
			if !ok {
				// 通道关闭（拉取协程退出），提交剩余消息
				log.Printf("\n 工作协程%d：消息通道关闭", id)
				if len(batch) > 0 {
					log.Printf("工作协程%d：通道关闭前提交剩余%d条消息", id, len(batch))
					c.commitBatch(id, batch)
				}
				return
			}

			log.Printf("\n 工作协程%d：收到消息，开始处理", id)
			log.Printf("   偏移量: %d, 消息Key: %s", msg.Offset, string(msg.Key))

			// 处理单条消息（带幂等性保障+状态更新）
			if err := processMessage(c.ctx, msg); err != nil {
				log.Printf(" 工作协程%d：消息处理失败（key=%s）", id, string(msg.Key))
				log.Printf("   错误原因: %v", err)
				_ = sendToDeadLetter(c.ctx, msg, err.Error())
				continue // 失败消息不加入提交批次
			}

			// 处理成功，加入批量提交队列
			batch = append(batch, msg)
			log.Printf(" 工作协程%d：消息处理成功（key=%s），已加入批量队列", id, string(msg.Key))
			log.Printf("   当前批量队列大小: %d/%d", len(batch), batchSize)

			// 达到批量阈值，触发提交
			if len(batch) >= batchSize {
				log.Printf("\n 工作协程%d：达到批量阈值（%d条），开始提交", id, batchSize)
				c.commitBatch(id, batch)
				batch = batch[:0]          // 清空批次
				ticker.Reset(batchTimeout) // 重置超时定时器
				log.Printf("工作协程%d：批量提交完成，等待新消息...\n", id)
			}

		case <-ticker.C:
			// 超时未达阈值，也提交（避免消息长时间未提交）
			if len(batch) > 0 {
				log.Printf("\n 工作协程%d：批量超时（%v），提交剩余%d条消息", id, batchTimeout, len(batch))
				c.commitBatch(id, batch)
				batch = batch[:0]
				log.Printf("工作协程%d：超时提交完成，等待新消息...\n", id)
			}
		}
	}
}

// 批量提交偏移量（增强日志）
func (c *AsyncConsumer) commitBatch(workerID int, msgs []kafka.Message) {
	// 记录提交的消息Key列表，便于排查
	msgKeys := make([]string, 0, len(msgs))
	for _, msg := range msgs {
		msgKeys = append(msgKeys, string(msg.Key))
	}

	if err := c.reader.CommitMessages(c.ctx, msgs...); err != nil {
		log.Printf(" 工作协程%d：批量提交失败", workerID)
		log.Printf("   失败消息Keys: %v", msgKeys)
		log.Printf("   错误原因: %v\n", err)
	} else {
		log.Printf(" 工作协程%d：批量提交成功", workerID)
		log.Printf("   提交消息数量: %d条", len(msgs))
		log.Printf("   提交消息Keys: %v", msgKeys)
		log.Printf("   最后偏移量: %d\n", msgs[len(msgs)-1].Offset)
	}
}

// 消息处理（新增订单状态更新+幂等保障）
func processMessage(ctx context.Context, msg kafka.Message) error {
	var orderMsg OrderMessage
	if err := json.Unmarshal(msg.Value, &orderMsg); err != nil {
		return fmt.Errorf("解析消息失败: %v", err)
	}

	log.Printf(" 消息解析成功：order_id=%s, amount=%d, 原始状态=%s",
		orderMsg.OrderID, orderMsg.Amount, orderMsg.Status)

	// 1. 幂等保障：避免重复入库（针对未通过本地消息表创建的订单）
	insertSQL := "INSERT IGNORE INTO orders (order_id, amount, status, create_time) VALUES (?, ?, ?, NOW())"
	_, err := common.SqlDB.ExecContext(ctx, insertSQL, orderMsg.OrderID, orderMsg.Amount, "pending")
	if err != nil {
		return fmt.Errorf("订单入库失败: %v", err)
	}

	// 2. 更新订单状态为"已支付"（核心闭环逻辑）
	updateSQL := `UPDATE orders 
	              SET status = 'paid', update_time = NOW() 
	              WHERE order_id = ? AND status = 'pending'` // 仅更新待支付状态

	result, err := common.SqlDB.ExecContext(ctx, updateSQL, orderMsg.OrderID)
	if err != nil {
		return fmt.Errorf("更新订单状态失败: %v", err)
	}

	// 3. 验证更新结果
	rowsAffected, err := result.RowsAffected()
	if err != nil {
		return fmt.Errorf("查询更新行数失败: %v", err)
	}
	if rowsAffected == 0 {
		return fmt.Errorf("未找到待更新订单（order_id=%s）或已处理", orderMsg.OrderID)
	}

	log.Printf(" 订单状态更新成功：order_id=%s, 旧状态=pending, 新状态=paid", orderMsg.OrderID)
	return nil
}

// 发送到死信队列（增强日志）
func sendToDeadLetter(ctx context.Context, msg kafka.Message, errMsg string) error {
	deadLetterTopic := "order_dead_letter"
	// 构造死信消息
	deadLetterValue := fmt.Sprintf(`{
		"original_key": "%s",
		"original_value": "%s",
		"error": "%s",
		"timestamp": %d
	}`, string(msg.Key), string(msg.Value), errMsg, time.Now().UnixMilli())

	log.Printf(" 发送死信消息：topic=%s, key=%s", deadLetterTopic, string(msg.Key))
	if err := kafkaUtil.SendKafkaMessage(ctx, deadLetterTopic, string(msg.Key), deadLetterValue); err != nil {
		log.Printf(" 发送死信队列失败（key=%s）: %v", string(msg.Key), err)
		return err
	}
	log.Printf(" 死信消息发送成功（key=%s）", string(msg.Key))
	return nil
}

// 停止消费者（优雅关闭）
func (c *AsyncConsumer) Stop() {
	log.Println("\n 开始关闭异步消费者...")
	c.cancel()           // 触发上下文取消
	c.wg.Wait()          // 等待所有工作协程退出
	_ = c.reader.Close() // 关闭 Kafka 读者
	log.Println(" 异步消费者已完全关闭")
}
